package com.buddy.zookeeper.worker;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.Random;

/**
 *
 */
public class Worker implements Watcher {

    private static final Logger LOG = LoggerFactory.getLogger(Worker.class);

    ZooKeeper zk;
    String hostPort;
    String serverId = Integer.toHexString(new Random().nextInt());

    public Worker(String hostPort) {
        this.hostPort = hostPort;
    }

    void startZK() throws IOException {
        zk = new ZooKeeper(hostPort,15000,this);
    }

    public void process(WatchedEvent event) {
        LOG.info(event.toString()+","+hostPort);
    }

    void register(){
        zk.create("/workers/worker-"+serverId,
                "Idle".getBytes(),
                ZooDefs.Ids.OPEN_ACL_UNSAFE,
                CreateMode.EPHEMERAL,
                createWorkerCallback,null);
    }

    AsyncCallback.StringCallback createWorkerCallback = new AsyncCallback.StringCallback() {
        public void processResult(int rc, String path, Object ctx, String name) {
            switch (KeeperException.Code.get(rc)){
                case CONNECTIONLOSS:
                    register();
                    break;
                case OK:
                    LOG.info("Registered successfully:"+serverId);
                    break;
                case NODEEXISTS:
                    LOG.warn("Already registered:"+serverId);
                    break;
                default:
                    LOG.error("Smoething went wrong:"+KeeperException.create(KeeperException.Code.get(rc),path));
            }
        }
    };

    AsyncCallback.StatCallback statusUpdateCallback = new AsyncCallback.StatCallback() {
        public void processResult(int rc, String path, Object ctx, Stat stat) {
            switch (KeeperException.Code.get(rc)){
                case CONNECTIONLOSS:
                    updateStatus((String) ctx);
                    return;
            }
        }
    };
    String status;
    synchronized private void updateStatus(String status) {
        if (status == this.status) {
            zk.setData("/workers/worker-" + serverId,
                    status.getBytes(),
                    -1,
                    statusUpdateCallback,
                    status);
        }
    }

    public void setStatus(String status){
        this.status = status;
        updateStatus(status);
    }

    public static void main(String[] args) throws Exception{
        Worker w = new Worker("127.0.0.1:2181");
        w.startZK();
        w.register();
        Thread.sleep(30000);
    }
}
